package org.fin.minicat.startup;

import org.fin.minicat.container.Service;
import org.fin.minicat.request.RequestProcessor;

import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;

public class Connector {

  private final int port;

  private Service service;

  private CyclicBarrier cyclicBarrier;

  public Connector(int port) {
    this.port = port;
  }

  public void start() {
    try {
      ServerSocketChannel ssc = ServerSocketChannel.open();
      ssc.bind(new InetSocketAddress(port));
      ssc.configureBlocking(false);

      Selector selector = Selector.open();
      ssc.register(selector, SelectionKey.OP_ACCEPT);

      // 等待其他Connector准备好再启动监听
      cyclicBarrier.await();
      while (ssc.isOpen()) {
        // 一些NIO的操作...
        selector.select();
        Set<SelectionKey> keys = selector.selectedKeys();
        for (Iterator<SelectionKey> iterator = keys.iterator(); iterator.hasNext(); ) {
          SelectionKey key = iterator.next();
          if (key.isAcceptable()) {
            ServerSocketChannel channel = (ServerSocketChannel) key.channel();
            SocketChannel sc = channel.accept();
            sc.configureBlocking(false);
            sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
          }
          if (key.isReadable() && key.isWritable()) {
            SocketChannel sc = (SocketChannel) key.channel();
            sc.register(selector, key.interestOps() ^ SelectionKey.OP_READ ^ SelectionKey.OP_WRITE);

            // 创建请求处理器RequestProcessor，并交给线程池执行
            Executor executor = getExecutor();
            executor.execute(new RequestProcessor(sc, service));

          }

          iterator.remove();
        }

      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  public Executor getExecutor() {
    return service.getExecutor();
  }

  public int getPort() {
    return port;
  }

  public Service getService() {
    return service;
  }

  public void setService(Service service) {
    this.service = service;
  }

  public void setCyclicBarrier(CyclicBarrier cyclicBarrier) {
    this.cyclicBarrier = cyclicBarrier;
  }
}
